package com.tensai.study.java.newconcept.typereactiveprogram.reactor.processor;

import org.junit.Test;
import reactor.core.Disposable;
import reactor.core.publisher.*;

/**
 * @author Tensai
 * @see Sinks.Many
 */
public class FluxProcessorTest {

	/**
	 * UnicastProcessor只允许有一个订阅者
	 *
	 * <p>UnicastProcessor可以在多个线程中生产元素，即多个生产源同时生产、下发元素。</p>
	 * <p>为了保证元素在多线程的情况下有序的下发，UnicastProcessor利用SerializedSink进行控制。</p>
	 * <p/>
	 * <b>总结</b>
	 * <li>只要有一个生产者元素下发完毕，其他的生产者元素都会被抛弃。</li>
	 * <li>UnicastProcessor作为发布者，可以在多个线程中同时发布元素，但是只能和一个订阅者建立订阅关系。</li>
	 * <li>支持背压。</li>
	 */
	@Test
	public void unicastTest() {
		UnicastProcessor<Object> processor = UnicastProcessor.create();
		FluxSink<Object> sink = processor.sink();
		sink.next("aaa");
		sink.next("aaa");
		sink.next("aaa");
		sink.next("aaa");
		Flux.just("Hello", "World").subscribe(processor);
		processor.subscribe(System.out::println);
		// 若打开以下注释，则报错，因为UnicastProcessor只允许有一个订阅者
		// processor.subscribe(System.out::println);
		sink.next("aaa");
		sink.next("aaa");
		sink.next("aaa");
		sink.next("aaa");
	}

	/**
	 * <li>DirectProcessor可以接收n个Subscriber。</li>
	 * <li>DirectProcessor不支持背压，其内部也没有用于存储元素的数据结构。</li>
	 * <li>DirectProcessor作为生产者存在的时候，只要对应的订阅者中有一个元素的请求数量为0，
	 * 就会移除该订阅者，并同时发送一个错误事件。</li>
	 *
	 * <b>总结</b>
	 * <li>DirectProcessor可以和多个订阅者建立订阅关系。</li>
	 * <li>只要有一个数据源的元素下发结束，其他数据源的元素就会被抛弃。</li>
	 * <li>不支持背压，也无数据存储。</li>
	 */
	@Test
	public void directTest() {
		DirectProcessor<Object> directProcessor = DirectProcessor.create();
		FluxSink<Object> sink = directProcessor.sink();
		// 1. 这里的元素是不会输出的，因为此时还并没有订阅者，
		sink.next("000");
		// 2.订阅后，即可输出发布源发出的元素
		directProcessor.subscribe(System.out::println);
		sink.next("111");
		directProcessor.subscribe(System.out::println);
		sink.next("222");
		// 上游的源一旦发布元素结束，即调用对应的onComplete方法，将对应的subscribe改为TERMlNATED
		// 此时后面的元素都会被抛弃，因此333,444都不会被输出
		Flux.just("Hello", "DockerX").subscribe(directProcessor);
		sink.next("333");
		sink.next("444");
		directProcessor.subscribe(System.out::println);
		sink.next("555");
	}

	/**
	 * <li>EmitterProcessor可以有多个订阅者，并提供背压支持。</li>
	 * <li>EmitterProcessor可以作为订阅者订阅上游，同步向下发送元素给自己的订阅者。</li>
	 *
	 * <b>总结</b>
	 * <li>EmitterProcessor作为源可以有多个订阅者，并且在没有订阅的情况下，可以存储上游的元素到队列中，并且不会被重复消费（被第一个订阅者给获取到），提供背压。</li>
	 * <li>EmitterProcessor在订阅关系数为0的时候，会自动清理队列中的缓存。</li>
	 * <li>EmitterProcessor作为订阅者不能同时订阅两个数据源。</li>
	 */
	@Test
	public void emitterTest() {
		EmitterProcessor<Object> emitterProcessor = EmitterProcessor.create();
		FluxSink<Object> sink = emitterProcessor.sink();
		// 1.此时没有订阅者，但是会将这个元素存储到队列中
		sink.next("000");
		// 2.此时有了订阅者，会将队列中的元素获取到，也因此000可以获取，但是如果用DirectProcessor就不能消费。
		// 同时保证该元素不会被重复消费，只有第一个订阅者能够取到
		Disposable s1 = emitterProcessor.subscribe(System.out::println);
		// 发布了元素xxx，此时订阅者s1已经产生订阅关系，可以输出元素,1条数据
		sink.next("xxx");
		Disposable s2 = emitterProcessor.subscribe(System.out::println);
		// 发布了元素aaa，此时订阅者s1和s2已经产生订阅关系，可以输出元素，因此有2条数据
		sink.next("aaa");
		// 这里会提示，一个订阅者不能同时订阅两个生产源
		Flux.just("Hello", "DockerX").subscribe(emitterProcessor);
		// 发布了元素bbb，此时订阅者s1和s2已经产生订阅关系，可以输出元素，因此有2条数据
		sink.next("bbb");
		Disposable s3 = emitterProcessor.subscribe(System.out::println);
		// 发布了元素ccc，此时订阅者s1、s2和s3已经产生订阅关系，可以输出元素，因此有3条数据
		sink.next("ccc");
		s1.dispose();// 关闭
		s2.dispose();// 关闭
		s3.dispose();// 关闭
		// 订阅关系已经取消，因此下面的ddd不能够获取到
		sink.next("ddd");
	}


}
